package com.aotain.project.apollo.bolt;

import java.text.SimpleDateFormat;
//import java.util.Calendar;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
//import java.util.Map.Entry;



//import org.apache.commons.codec.binary.Base64;
import org.apache.log4j.Logger;
import org.apache.storm.Config;
import org.apache.storm.Constants;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.FailedException;
import org.apache.storm.topology.IRichBolt;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Tuple;

import com.aotain.hbase.HBaseRecordAdd;
import com.aotain.apollo.ApolloConfig;
import com.aotain.storm.AbstractBolt;

/**
 * 根据 域名和 机房统计 PV值
 * 2016/02/23 新建
 * @author 胡德意
 *
 */
public class IDCStatisBolt extends AbstractBolt{

	/**
	 * 根域名PV统计
	 */
	
	
	OutputCollector _collector;
	
	String IDCStatMapKey = null;
	Map<String,Long> IDCStatMap = new HashMap<String,Long>();
	
	
	private void countInLocal(Tuple tuple) {

		if(tuple != null ) {  
			//System.out.println("..........IDCStatisBolt......... " + System.currentTimeMillis());

			String IDC = tuple.getStringByField("IDC");
			String TYPE = tuple.getStringByField("TYPE");
			long IDCStatPV = tuple.getLongByField("PV");
			if(IDC==null || IDC.trim().isEmpty()||TYPE==null || TYPE.trim().isEmpty())
				return;	
			
			IDCStatMapKey = IDC + "*"  + TYPE;			
			if(IDCStatMap.get(IDCStatMapKey) != null) {
				IDCStatPV = IDCStatMap.get(IDCStatMapKey) + IDCStatPV;
			}
			//System.out.println("IDC is " + IDC +",TYPE is " + TYPE +", pv is "+ IDCStatPV);
			IDCStatMap.put(IDCStatMapKey, IDCStatPV);			
			
		} else {  
			System.err.println(" >>>>>>>>>>>>>>>>IDCStatisBolt tuple is null <<<<<<<<<<<<<<<");
		}
		
		//TotalPV++;
	}
	
	private void emitCountingData() {
		
		SimpleDateFormat sdfm = new SimpleDateFormat("mm");
		SimpleDateFormat sdfh = new SimpleDateFormat("yyyyMMddHH");
		SimpleDateFormat sdfd = new SimpleDateFormat("yyyyMMdd");
		Date date = new Date();
		String dateStrM = sdfm.format(date);
		String dateStrH = sdfh.format(date);
		String dateStrD = sdfd.format(date);
		int nMin5 = (Integer.parseInt(dateStrM)/5)*5;
		int nMin30 = (Integer.parseInt(dateStrM)/30)*30;
		String sMin5 = String.format("%02d", nMin5);
		String sMin30 = String.format("%02d", nMin30);
		String dateStr5 = dateStrH+sMin5+"00";
		String dateStr30 = dateStrH+sMin30+"00";
		String rowKeyIDCStat5Min = null;
		String rowKeyIDCStat30Min = null;
		String rowKeyIDCStatH = null;
		String rowKeyIDCStatD = null;
	
		HBaseRecordAdd hbaseInstance = HBaseRecordAdd.getInstance(
				zooserver, logserver, brokerList);
		
		for(String dkey : IDCStatMap.keySet()) {
			
			long IDCStatPV = IDCStatMap.get(dkey);	
			String[] splits3 = dkey.split("\\*",-1);
			String IDC=splits3[0];
			rowKeyIDCStat5Min = IDC + "_" + dateStr5;
			rowKeyIDCStat30Min = IDC + "_" + dateStr30;
			rowKeyIDCStatH =   IDC + "_" + dateStrH;
			rowKeyIDCStatD =   IDC + "_" + dateStrD;
			String type = splits3[1];	
			String cf = "cf";
			
			//System.out.println("IDCStat30Min is " + rowKeyIDCStat30Min +",IDCStatD is " + rowKeyIDCStatD +",IDCStat5Min is " + rowKeyIDCStat5Min +",TYPE is " + type +" is "+ IDCStatPV);
			
			
			hbaseInstance.Add("SDS_IDC_STAT_5MIN", rowKeyIDCStat5Min, cf, "IDC", IDC);
			hbaseInstance.Add("SDS_IDC_STAT_30MIN", rowKeyIDCStat30Min, cf, "IDC", IDC);
			hbaseInstance.Add("SDS_IDC_STAT_H", rowKeyIDCStatH, cf, "IDC", IDC);
			hbaseInstance.Add("SDS_IDC_STAT_D", rowKeyIDCStatD, cf, "IDC", IDC);

			hbaseInstance.Add("SDS_IDC_STAT_5MIN", rowKeyIDCStat5Min, cf, "REPORTTIME", dateStr5);
			hbaseInstance.Add("SDS_IDC_STAT_30MIN", rowKeyIDCStat30Min, cf, "REPORTTIME", dateStr30);
			hbaseInstance.Add("SDS_IDC_STAT_H", rowKeyIDCStatH, cf, "REPORTTIME", dateStrH+"0000");
			hbaseInstance.Add("SDS_IDC_STAT_D", rowKeyIDCStatD, cf, "REPORTTIME", dateStrD);
			
			hbaseInstance.incerment("SDS_IDC_STAT_5MIN", rowKeyIDCStat5Min, "cf:"+type, IDCStatPV);
			hbaseInstance.incerment("SDS_IDC_STAT_30MIN", rowKeyIDCStat30Min, "cf:"+type, IDCStatPV);
			hbaseInstance.incerment("SDS_IDC_STAT_H", rowKeyIDCStatH, "cf:"+type, IDCStatPV);
			hbaseInstance.incerment("SDS_IDC_STAT_D", rowKeyIDCStatD, "cf:"+type, IDCStatPV);


			hbaseInstance.Add("SDS_IDC_STAT_5MIN", dateStr5, cf, "REPORTTIME", dateStr5);
			hbaseInstance.Add("SDS_IDC_STAT_30MIN", dateStr30, cf, "REPORTTIME", dateStr30);
			hbaseInstance.Add("SDS_IDC_STAT_H", dateStrH, cf, "REPORTTIME", dateStrH+"0000");
			hbaseInstance.Add("SDS_IDC_STAT_D", dateStrD, cf, "REPORTTIME", dateStrD);
			
			hbaseInstance.incerment("SDS_IDC_STAT_5MIN", dateStr5, "cf:"+type, IDCStatPV);
			hbaseInstance.incerment("SDS_IDC_STAT_30MIN", dateStr30, "cf:"+type, IDCStatPV);
			hbaseInstance.incerment("SDS_IDC_STAT_H", dateStrH, "cf:"+type, IDCStatPV);
			hbaseInstance.incerment("SDS_IDC_STAT_D", dateStrD, "cf:"+type, IDCStatPV);
			
			//System.out.println("IDCStat30Min is " + dateStr30 +",IDCStatD is " + dateStrD +",IDCStat5Min is " + dateStr5 +",TYPE is " + type +" is "+ IDCStatPV);
			
			
		}
		
		IDCStatMap.clear();
		
	}
	
	@Override
	public void execute(Tuple input) {
		// TODO Auto-generated method stub
		try {
			
			if(isTickTuple(input)) {//满足时间条件
				emitCountingData();  //计算，入库
				_collector.ack(input);
			} else {//不满足时间条件
				countInLocal(input); //数据记录 
				_collector.ack(input);
			}
		} catch (Exception e) {
			// TODO Auto-generated catch block
			Logger.getRootLogger().error(e);
			throw new FailedException("IDCStatisBolt Exception",e);
		}
	}

	@Override
	public void cleanup() {
		// TODO Auto-generated method stub
		
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		// TODO Auto-generated method stub
		
	}

	@Override
	public Map<String, Object> getComponentConfiguration() {
		// TODO Auto-generated method stub
		Map<String,Object> conf = new HashMap<String,Object>();
		conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS,300);
		return conf;
	}
	

	@Override
	public void Init(Map stormConf, TopologyContext context,
			OutputCollector collector) {
		// TODO Auto-generated method stub
		_collector = collector;
		String dbJson = stormConf.get("dbjson").toString();
		ApolloConfig ap = new ApolloConfig(dbJson);
	}

}
